

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Collections.Generic;
    using System.Text;
    using Confluent.Kafka;
    using NLog;
    using Stee.CAP8.MessageBus.Interfaces.Delegates;
    using Stee.CAP8.MessageBus.Interfaces.Models;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Models;

    /// <summary>
    /// Poll consumption implementation for event messages.
    /// Event messages uses priority levels to process messages as major events like incident update
    /// may potentially have a high number of listening clients.
    /// </summary>
    internal sealed class PollingEventImpl : PollingBase
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        #endregion Constant & Static Variables


        #region Fields
        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the subscription handler for event messages
        /// </summary>
        public HandlerEventDetails SubscribeHandler { get; private set; }

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the PollingEventImpl class
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="groupID">Subscriber group ID</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        internal PollingEventImpl(string brokerConfigurationJSON, string groupID, HandleBrokerError brokerErrorHandler)
            : base(brokerConfigurationJSON, groupID, brokerErrorHandler)
        {
            this.SubscribeHandler = new HandlerEventDetails();
        }

        /// <summary>
        /// Destructor
        /// </summary>
        ~PollingEventImpl()
        {
            this.Dispose(true);
        }

        #endregion Constructors & Initializers


        #region Public Methods
        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Handle consumed messsage
        /// </summary>
        /// <param name="consumeResult">Raw message from Kafka broker</param>
        /// <param name="properties">Message meta-data</param>
        /// <returns>Return TRUE if handled successfully else FALSE</returns>
        protected override bool HandleMessage(ConsumeResult<string, byte[]> consumeResult, IDictionary<string, byte[]> properties)
        {
            //// Extract header for main message type
            var messageType = string.Empty;
            var isHandled = false;

            try
            {
                if (consumeResult.Message.Headers.TryGetLastBytes(HeaderKeys.HeaderKeyMessageType, out var messageTypeHeader))
                {
                    messageType = DeserializeString(messageTypeHeader, Encoding.UTF8);
                }

                if (string.Equals(messageType, HeaderKeys.MessageTypeEvent, StringComparison.InvariantCultureIgnoreCase))
                {
                    var utcDT = consumeResult.Message.Timestamp.UtcDateTime;
                    var message = EventMessage.Generate(consumeResult.Message.Value, properties);

                    //// Invoke handler
                    isHandled = this.SubscribeHandler.Invoke(this.TopicName, message, utcDT);
                }
                else
                {
                    //// Message type is not of event message type
                    Logger.Error($"Failed to handle message, message type is not of event type. Type: {messageType}");
                    isHandled = true;
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }

            return isHandled;
        }

        #endregion Private Methods


        #region Static Methods
        #endregion Static Methods
    }
}
